package com.lhd.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lhd.bean.TrafficPageViewBean;
import com.lhd.common.utils.MyClickHouseUtil;
import com.lhd.common.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class DwsTrafficVcChArlsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> uvDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_unique_visitor_detail", "vccharisnew_pageview_window_lhd"));
        DataStreamSource<String> ujDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_user_jump_detail", "vccharisnew_pageview_window_lhd"));
        DataStreamSource<String> pageDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "vccharisnew_pageview_window_lhd"));

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUvDs = uvDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");

            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L, 0L, 0L, 0L, 0L,
                    jsonObject.getLong("ts"));
        });

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithUjDS = ujDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");

            return new TrafficPageViewBean(null, null,
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 0L, 0L, 0L, 1L,
                    jsonObject.getLong("ts"));
        });

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithPageDS = pageDs.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            JSONObject common = jsonObject.getJSONObject("common");
            JSONObject page = jsonObject.getJSONObject("page");

            String lastPageId = page.getString("last_page_id");
            long sv = 0L;//回话数
            if (lastPageId == null) {
                sv = 1L;
            }

            return new TrafficPageViewBean(null, null,
                    common.getString("vc"), //版本
                    common.getString("ch"), //驱动
                    common.getString("ar"), //区域
                    common.getString("is_new"), //新老用户
                    0L, //独立
                    sv,//回话
                    1L,
                    page.getLong("during_time"),
                    0L,
                    jsonObject.getLong("ts"));
        });

        trafficPageViewWithUjDS.print("uj===>");
        trafficPageViewWithUvDs.print("uv===>");
        trafficPageViewWithPageDS.print("page===>");

        DataStream<TrafficPageViewBean> unionDS = trafficPageViewWithUvDs.union(
                trafficPageViewWithUjDS,
                trafficPageViewWithPageDS
        );

        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewWithWmDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        trafficPageViewWithWmDS.print("查看水位线------>>>>");
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowedStream = trafficPageViewWithWmDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {
                        return new Tuple4<>(value.getAr(),
                                value.getCh(),
                                value.getIsNew(),
                                value.getVc());
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .allowedLateness(Time.seconds(10l));

        SingleOutputStreamOperator<TrafficPageViewBean> resultDS = windowedStream.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                return value1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {

            @Override
            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                TrafficPageViewBean next = input.iterator().next();

                next.setStt(new Timestamp(window.getStart()));
                next.setEdt(new Timestamp(window.getEnd()));
                next.setTs(System.currentTimeMillis());
                out.collect(next);
            }
        });
        resultDS.print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
        resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into gmall.dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));


        env.execute();
    }
}
